package com.niodata.dt.fs;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.ftp.FTPFileSystem;

public class FtpFileSystemBuilder implements FileSystemBuilder {

  @Override
  public FileSystem createFileSystem(String user, Map<String, String> conf) throws IOException {
    Configuration config = new Configuration();
    String host = conf.get("fs.ftp.host");
    String port = conf.getOrDefault("fs.ftp.host.port", "21");
    String ftpUser = conf.getOrDefault("fs.ftp.user", "anonymous");
    String ftpPassword = conf.getOrDefault("fs.ftp.password", "anonymous");

    String defaultFs = String.format("ftp://%s@%s:%s", ftpUser, host, port);
    config.set("fs.defaultFS", defaultFs);
    config.setClass("fs.ftp.impl", DpFtpFileSystem.class, FileSystem.class);
    config.set(FTPFileSystem.FS_FTP_HOST, host);
    config.set(FTPFileSystem.FS_FTP_HOST_PORT, port);
    config.set(FTPFileSystem.FS_FTP_USER_PREFIX + host, ftpUser);
    config.set(FTPFileSystem.FS_FTP_PASSWORD_PREFIX + host, ftpPassword);
    config.set(DpFtpFileSystem.FS_FTP_PROXY_HOST,
          conf.getOrDefault(DpFtpFileSystem.FS_FTP_PROXY_HOST, ""));
    config.set(DpFtpFileSystem.FS_FTP_PROXY_PORT,
          conf.getOrDefault(DpFtpFileSystem.FS_FTP_PROXY_PORT, ""));
    config.set(DpFtpFileSystem.FS_FTP_PROXY_USER,
          conf.getOrDefault(DpFtpFileSystem.FS_FTP_PROXY_USER, ""));
    config.set(DpFtpFileSystem.FS_FTP_PROXY_PASSWORD,
          conf.getOrDefault(DpFtpFileSystem.FS_FTP_PROXY_PASSWORD, ""));
    config.set(DpFtpFileSystem.FS_FTP_TIMEZONE,
          conf.getOrDefault(DpFtpFileSystem.FS_FTP_TIMEZONE, "UTC"));
    FileSystem fs = FileSystem.get(config);
    return fs;
  }
}
